package cn.lsh.kafka.consumer;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.time.LocalTime;
import java.util.Arrays;
import java.util.Map;
import java.util.Properties;

public class FirstConsumer {

	public static void main(String[] args) {
		Properties properties = new Properties();
		properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "node00:9092,node01:9092,node02:9092");
		properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.IntegerDeserializer");
		properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
		//自动提交消费者偏移量,默认为true；如果客户端消费了消息没有提交offset，下次重启后还是会从原来的偏移量开始消费，即会重复消费；因为重启会重新拉取offset，而不是访问客户端内存中的offset
		properties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
		//自动提交偏移量的延时时间，单位毫秒，ENABLE_AUTO_COMMIT_CONFIG必须配置为true
		properties.setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
		//配置消费者组
		properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "group_1");
		//配置是否重置offerset，earliest：表示启动时从头开始消费，客户端要配置不用的消费者组,；latest：表示从最新开始消费，默认使用
		properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
		//消费者一批次拉取的最大消息数
		properties.setProperty(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "5");
		//设置消费端分区策略，默认为org.apache.kafka.clients.consumer.RangeAssignor
		// properties.setProperty(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, "org.apache.kafka.clients.consumer.RoundRobinAssignor");
		properties.setProperty(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, "cn.lsh.kafka.partitioner.MyConsumerAssignor");
		//是否排除kafka内部主题，默认为true，即排除默认主题
		properties.setProperty(ConsumerConfig.EXCLUDE_INTERNAL_TOPICS_CONFIG, "true");


		KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(properties);
		// consumer.subscribe(Collections.singleton("__comsumer_offsets"));
		consumer.subscribe(Arrays.asList("ddd-1"));

		while (true) {
			//poll拉取数据，根据传入的时间阻塞当前线程
			ConsumerRecords<Integer, String> records = consumer.poll(Duration.ofMillis(10000));
			if (records.count() > 0) {
				System.out.println("当前批次拉取数据：" + records.count() + "，时间：" + LocalTime.now());
			}
			for (ConsumerRecord<Integer, String> record : records) {
				System.out.println(record.offset() + "     " + record.key() + "     " + record.value());
			}
			//同步提交，如果提交一直不成功，则会一直重试，阻塞当前进程
			// consumer.commitSync();
			//异步提交
			consumer.commitAsync((map, e) -> {
				if (e == null) {
					System.out.println("offset提交成功");
				} else {
					System.out.println("offset提交失败");
				}
			});
		}

	}

}
